package com.ymm.spark.spark.session;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Optional;
import com.ymm.spark.config.ConfigurationManager;
import com.ymm.spark.constant.Constants;
import com.ymm.spark.dao.ISessionAggrStatDAO;
import com.ymm.spark.dao.ISessionDetailDAO;
import com.ymm.spark.dao.ISessionRandomExtractDAO;
import com.ymm.spark.dao.ITaskDAO;
import com.ymm.spark.dao.ITop10CategoryDAO;
import com.ymm.spark.dao.ITop10SessionDAO;
import com.ymm.spark.dao.factory.DAOFactory;
import com.ymm.spark.domain.SessionAggrStat;
import com.ymm.spark.domain.SessionDetail;
import com.ymm.spark.domain.SessionRandomExtract;
import com.ymm.spark.domain.Task;
import com.ymm.spark.domain.Top10Category;
import com.ymm.spark.domain.Top10Session;
import com.ymm.spark.test.MockData;
import com.ymm.spark.util.DateUtils;
import com.ymm.spark.util.NumberUtils;
import com.ymm.spark.util.ParamUtils;
import com.ymm.spark.util.StringUtils;
import com.ymm.spark.util.ValidUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * @author zhangj
 * @date 2019/11/28
 */
public class UserSessionTask {
	public static void main(String[] args) {
		args = new String[]{"1"};
		//构建spark上下文
		SparkConf conf = new SparkConf()
				.setAppName("UserSessionTask")
				.setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);

		SQLContext sqlContext = new SQLContext(sc);
		//生成模拟数据
		mockData(sc, sqlContext);

		//获取taskDAO
		ITaskDAO taskDAO = DAOFactory.getTaskDAO();

		//获取taskid,taskid是从J2EE平台执行spark-submit shell脚本,传递给args参数的
		Long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PAGE);

		//获取task信息
		Task task = taskDAO.findById(taskid);

		//获取要执行任务的查询参数
		JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());


		/**
		 * 根据条件从user_visit_action Hive表中过滤需要的数据,actionRDD可以作为原始数据
		 * [2019-11-28,80,1ad5408d88f047d1bbe89e3d09dab8a8,7,2019-11-28 9:35:01,呷哺呷哺,null,null,null,null,null,null]
		 * [2019-11-28,80,1ad5408d88f047d1bbe89e3d09dab8a8,8,2019-11-28 9:11:56,null,null,null,94,13,null,null]
		 * [2019-11-28,80,1ad5408d88f047d1bbe89e3d09dab8a8,9,2019-11-28 9:44:11,蛋糕,null,null,null,null,null,null]
		 * [2019-11-28,80,1ad5408d88f047d1bbe89e3d09dab8a8,6,2019-11-28 9:05:43,null,null,null,null,null,82,86]
		 */
		JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);

		/**
		 * 转换成<sessionid,Row>的格式
		 * 1ad5408d88f047d1bbe89e3d09dab8a8 /
		 * 2019-11-28,80,1ad5408d88f047d1bbe89e3d09dab8a8,7,2019-11-28 9:35:01,呷哺呷哺,null,null,null,null,null,null
		 */
		JavaPairRDD<String, Row> sessionid2actionRDD = getSessionid2ActionRDD(actionRDD);


		/**
		 * 转换成<sessionid,(sessionid,searchKeywords,clickCategoryIds,age,professional,city,sex)>
		 */
		JavaPairRDD<String, String> sessionid2AggrInfoRDD = aggregateBySession(sqlContext, sessionid2actionRDD);

		/**
		 * 根据传来的条件进行过滤并统计聚合数据
		 */
		//使用Accumulator进行自定义统计计算访问时长和步长
		Accumulator<String> sessionAggrStatAccumulator = sc.accumulator("", new SessionAggrStatAcumulator());
		JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = filterSessionAndAggrStat(sessionid2AggrInfoRDD, taskParam, sessionAggrStatAccumulator);

		/**
		 * 生产公共的RDD,通过筛选条件的sessionid的访问明细
		 */
		JavaPairRDD<String, Row> sessionid2detailRDD = getSessionid2detailRDD(filteredSessionid2AggrInfoRDD, sessionid2actionRDD);

		/**
		 * 随机抽取模块
		 */
		randomExtractSession(taskid, filteredSessionid2AggrInfoRDD, sessionid2actionRDD);

		/**
		 * 计算出各个范围的session占比,并写入Mysql,Accumulator只有在action之后才生效
		 */
		calculateAggrStat(sessionAggrStatAccumulator.value(), task.getTaskid());

		/**
		 * 获取top10热门品类
		 */
		List<Tuple2<CategorySortKey, String>> top10CategoryList = getTop10Category(task.getTaskid(), sessionid2detailRDD);

		/**
		 * 获取top10活跃session
		 */
		getTop10Session(sc, task.getTaskid(), top10CategoryList, sessionid2detailRDD);

		/**
		 * 关闭Spark上下文
		 */
		sc.close();
	}

	private static void getTop10Session(JavaSparkContext sc, final long taskid, List<Tuple2<CategorySortKey, String>> top10CategoryList,
	                                    JavaPairRDD<String, Row> sessionid2detailRDD) {
		/**
		 * 第一步:将top10热门品类的id,生成一份RDD
		 */
		ArrayList<Tuple2<Long, Long>> top10CategoryIdList = new ArrayList<Tuple2<Long, Long>>();
		for (Tuple2<CategorySortKey, String> category : top10CategoryList) {
			long categoryid = Long.valueOf(StringUtils.getFieldFromConcatString(
					category._2, "\\|", Constants.FIELD_CATEGORY_ID));
			top10CategoryIdList.add(new Tuple2<Long, Long>(categoryid, categoryid));
		}

		JavaPairRDD<Long, Long> top10CategoryIdRDD = sc.parallelizePairs(top10CategoryIdList);

		/**
		 * 第二步:计算top10品类被各session点击的次数
		 */
		JavaPairRDD<String, Iterable<Row>> sessionid2detailsRDD = sessionid2detailRDD.groupByKey();

		JavaPairRDD<Long, String> categoryid2sessionCountRDD = sessionid2detailsRDD.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
					@Override
					public Iterable<Tuple2<Long, String>> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
						String sessionid = tuple._1;
						Iterator<Row> iterator = tuple._2.iterator();
						HashMap<Long, Long> categoryCountMap = new HashMap<Long, Long>();

						while (iterator.hasNext()) {
							Row row = iterator.next();
							if (row.get(6) != null) {
								long categoryid = row.getLong(6);
								Long count = categoryCountMap.get(categoryid);
								if (count == null) {
									count = 0L;
								}
								count++;
								categoryCountMap.put(categoryid, count);
							}
						}
						//返回<categoryid,sessionid,count>
						ArrayList<Tuple2<Long, String>> list = new ArrayList<Tuple2<Long, String>>();

						for (Map.Entry<Long, Long> categoryCountEntry : categoryCountMap.entrySet()) {
							long categoryid = categoryCountEntry.getKey();
							long count = categoryCountEntry.getValue();
							String value = sessionid + "," + count;
							list.add(new Tuple2<Long, String>(categoryid, value));
						}

						return list;
					}
				});

		//获取到to10热门品类，被各个session点击的次数
		JavaPairRDD<Long, String> top10CategorySessionCountRDD =
				top10CategoryIdRDD.join(categoryid2sessionCountRDD)
						.mapToPair(new PairFunction<Tuple2<Long, Tuple2<Long, String>>, Long, String>() {
							@Override
							public Tuple2<Long, String> call(Tuple2<Long, Tuple2<Long, String>> tuple) throws Exception {
								return new Tuple2<Long, String>(tuple._1, tuple._2._2);
							}
						});

		/**
		 * 第三步:分组取topN算法,获取每个品类的top10活跃用户
		 */
		JavaPairRDD<Long, Iterable<String>> top10CategorySessionCountsRDD = top10CategorySessionCountRDD.groupByKey();

		JavaPairRDD<String, String> top10SessionRDD = top10CategorySessionCountsRDD.flatMapToPair(

				new PairFlatMapFunction<Tuple2<Long, Iterable<String>>, String, String>() {

					private static final long serialVersionUID = 1L;

					@Override
					public Iterable<Tuple2<String, String>> call(
							Tuple2<Long, Iterable<String>> tuple)
							throws Exception {
						long categoryid = tuple._1;
						Iterator<String> iterator = tuple._2.iterator();

						// 定义取topn的排序数组
						String[] top10Sessions = new String[10];

						while (iterator.hasNext()) {
							String sessionCount = iterator.next();
							long count = Long.valueOf(sessionCount.split(",")[1]);

							// 遍历排序数组
							for (int i = 0; i < top10Sessions.length; i++) {
								// 如果当前i位，没有数据，那么直接将i位数据赋值为当前sessionCount
								if (top10Sessions[i] == null) {
									top10Sessions[i] = sessionCount;
									break;
								} else {
									long _count = Long.valueOf(top10Sessions[i].split(",")[1]);

									// 如果sessionCount比i位的sessionCount要大
									if (count > _count) {
										// 从排序数组最后一位开始，到i位，所有数据往后挪一位
										for (int j = 9; j > i; j--) {
											top10Sessions[j] = top10Sessions[j - 1];
										}
										// 将i位赋值为sessionCount
										top10Sessions[i] = sessionCount;
										break;
									}

									// 比较小，继续外层for循环
								}
							}
						}

						// 将数据写入MySQL表
						List<Tuple2<String, String>> list = new ArrayList<Tuple2<String, String>>();

						for (String sessionCount : top10Sessions) {
							String sessionid = sessionCount.split(",")[0];
							long count = Long.valueOf(sessionCount.split(",")[1]);

							// 将top10 session插入MySQL表
							Top10Session top10Session = new Top10Session();
							top10Session.setTaskid(taskid);
							top10Session.setCategoryid(categoryid);
							top10Session.setSessionid(sessionid);
							top10Session.setClickCount(count);

							ITop10SessionDAO top10SessionDAO = DAOFactory.getITop10SessionDAO();
							top10SessionDAO.insert(top10Session);

							// 放入list
							list.add(new Tuple2<String, String>(sessionid, sessionid));
						}

						return list;
					}

				});

		/**
		 * 第四步：获取top10活跃session的明细数据，并写入MySQL
		 */
		JavaPairRDD<String, Tuple2<String, Row>> sessionDetailRDD =
				top10SessionRDD.join(sessionid2detailRDD);
		sessionDetailRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
				Row row = tuple._2._2;

				SessionDetail sessionDetail = new SessionDetail();
				sessionDetail.setTaskid(taskid);
				sessionDetail.setUserid(row.getLong(1));
				sessionDetail.setSessionid(row.getString(2));
				sessionDetail.setPageid(row.getLong(3));
				sessionDetail.setActionTime(row.getString(4));
				sessionDetail.setSearchKeyword(row.getString(5));
				sessionDetail.setClickCategoryId(row.getLong(6));
				sessionDetail.setClickProductId(row.getLong(7));
				sessionDetail.setOrderCategoryIds(row.getString(8));
				sessionDetail.setOrderProductIds(row.getString(9));
				sessionDetail.setPayCategoryIds(row.getString(10));
				sessionDetail.setPayProductIds(row.getString(11));

				ISessionDetailDAO sessionDetailDAO = DAOFactory.getISessionDetailDAO();
				sessionDetailDAO.insert(sessionDetail);
			}
		});
	}

	private static List<Tuple2<CategorySortKey, String>> getTop10Category(long taskid, JavaPairRDD<String, Row> sessionid2detailRDD) {
		/**
		 * 第一步:获取符合条件的session访问过的所有品类
		 */
		//获取session访问过的所有品类id,访问过指的是点击过,下单过,支付过的品类
		JavaPairRDD<Long, Long> categoryidRDD = sessionid2detailRDD.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
					@Override
					public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
						Row row = tuple._2;

						ArrayList<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();

						Long clickCategoryId = row.getLong(6);
						if (clickCategoryId != null) {
							list.add(new Tuple2<Long, Long>(clickCategoryId, clickCategoryId));
						}

						String orderCategoryIds = row.getString(8);
						if (orderCategoryIds != null) {
							String[] orderCategoryIdsSplited = orderCategoryIds.split(",");
							for (String orderCategoryId : orderCategoryIdsSplited) {
								list.add(new Tuple2<Long, Long>(Long.valueOf(orderCategoryId), Long.valueOf(orderCategoryId)));
							}
						}

						String payCategoryIds = row.getString(10);
						if (payCategoryIds != null) {
							String[] payCategoryIdsSplited = payCategoryIds.split(",");
							for (String payCategoryId : payCategoryIdsSplited) {
								list.add(new Tuple2<Long, Long>(Long.valueOf(payCategoryId),
										Long.valueOf(payCategoryId)));
							}
						}
						return list;
					}
				});

		//去重
		categoryidRDD = categoryidRDD.distinct();

		/**
		 * 第二步:计算各品类的点击,下单,和支付此时
		 */
		//计算各个品类的点击次数
		JavaPairRDD<Long, Long> clickCategoryId2CountRDD = getClickCategoryId2CountRDD(sessionid2detailRDD);
		// 计算各个品类的下单次数
		JavaPairRDD<Long, Long> orderCategoryId2CountRDD = getOrderCategoryId2CountRDD(sessionid2detailRDD);
		// 计算各个品类的支付次数
		JavaPairRDD<Long, Long> payCategoryId2CountRDD = getPayCategoryId2CountRDD(sessionid2detailRDD);

		/**
		 * 第三步:join各个品类与它的点击,下单,支付次数
		 */
		JavaPairRDD<Long, String> categoryid2countRDD = joinCategoryAndData(categoryidRDD, clickCategoryId2CountRDD,
				orderCategoryId2CountRDD, payCategoryId2CountRDD);

		/**
		 * 第四步:自定义二次排序
		 */

		/**
		 * 第五步:将数据映射成<CategorySortKey,info>格式的RDD，然后进行二次排序（降序）
		 */
		JavaPairRDD<CategorySortKey, String> sortKey2countRDD = categoryid2countRDD.mapToPair(
				new PairFunction<Tuple2<Long, String>, CategorySortKey, String>() {
					@Override
					public Tuple2<CategorySortKey, String> call(Tuple2<Long, String> tuple) throws Exception {
						String countInfo = tuple._2;

						long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
								countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
						long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
								countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
						long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
								countInfo, "\\|", Constants.FIELD_PAY_COUNT));
						CategorySortKey sortKey = new CategorySortKey(clickCount, orderCount, payCount);
						return new Tuple2<CategorySortKey, String>(sortKey, countInfo);
					}
				});
		JavaPairRDD<CategorySortKey, String> sortedCategoryCountRDD = sortKey2countRDD.sortByKey(false);

		/**
		 * 第六步:用take(10),取出top10热门品类,并写入mysql
		 */
		List<Tuple2<CategorySortKey, String>> top10CategoryList = sortedCategoryCountRDD.take(10);

		ITop10CategoryDAO top10CategoryDAO = DAOFactory.getITop10CategoryDAO();

		for (Tuple2<CategorySortKey, String> tuple : top10CategoryList) {
			String countInfo = tuple._2;
			long categoryid = Long.valueOf(StringUtils.getFieldFromConcatString(
					countInfo, "\\|", Constants.FIELD_CATEGORY_ID));
			long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
					countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
			long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
					countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
			long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
					countInfo, "\\|", Constants.FIELD_PAY_COUNT));

			Top10Category category = new Top10Category();
			category.setTaskid(taskid);
			category.setCategoryid(categoryid);
			category.setClickCount(clickCount);
			category.setOrderCount(orderCount);
			category.setPayCount(payCount);
			top10CategoryDAO.insert(category);
		}

		return top10CategoryList;
	}

	private static JavaPairRDD<Long, String> joinCategoryAndData(JavaPairRDD<Long, Long> categoryidRDD,
	                                                             JavaPairRDD<Long, Long> clickCategoryId2CountRDD,
	                                                             JavaPairRDD<Long, Long> orderCategoryId2CountRDD,
	                                                             JavaPairRDD<Long, Long> payCategoryId2CountRDD) {
		//leftOuterJoin()
		JavaPairRDD<Long, Tuple2<Long, Optional<Long>>> tmpJoinRDD = categoryidRDD.leftOuterJoin(clickCategoryId2CountRDD);

		JavaPairRDD<Long, String> tmpMapRDD = tmpJoinRDD.mapToPair(
				new PairFunction<Tuple2<Long, Tuple2<Long, Optional<Long>>>, Long, String>() {
					@Override
					public Tuple2<Long, String> call(Tuple2<Long, Tuple2<Long, Optional<Long>>> tuple) throws Exception {
						Long categoryid = tuple._1;
						Optional<Long> optional = tuple._2._2;
						long clickCount = 0L;
						if (optional.isPresent()) {
							clickCount = optional.get();
						}
						String value = Constants.FIELD_CATEGORY_ID + "=" + categoryid + "|" + Constants.FIELD_CLICK_COUNT + "=" + clickCount;
						return new Tuple2<Long, String>(categoryid, value);
					}
				});

		tmpMapRDD = tmpMapRDD.leftOuterJoin(orderCategoryId2CountRDD).mapToPair(
				new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
					@Override
					public Tuple2<Long, String> call(Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
						Long categoryid = tuple._1;
						String value = tuple._2._1;

						Optional<Long> optional = tuple._2._2;
						long orderCount = 0L;
						if (optional.isPresent()) {
							orderCount = optional.get();
						}
						value = value + "|" + Constants.FIELD_ORDER_COUNT + "=" + orderCount;
						return new Tuple2<Long, String>(categoryid, value);
					}
				});

		tmpMapRDD = tmpMapRDD.leftOuterJoin(payCategoryId2CountRDD).mapToPair(
				new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
					@Override
					public Tuple2<Long, String> call(Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
						long categoryid = tuple._1;
						String value = tuple._2._1;

						Optional<Long> optional = tuple._2._2;
						long payCount = 0L;

						if (optional.isPresent()) {
							payCount = optional.get();
						}
						value = value + "|" + Constants.FIELD_PAY_COUNT + "=" + payCount;
						return new Tuple2<Long, String>(categoryid, value);
					}
				});
		return tmpMapRDD;
	}

	private static JavaPairRDD<Long, Long> getPayCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD) {
		JavaPairRDD<String, Row> payActionRDD = sessionid2detailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
			@Override
			public Boolean call(Tuple2<String, Row> tuple) throws Exception {
				Row row = tuple._2;
				return row.getString(10) != null;
			}
		});
		JavaPairRDD<Long, Long> payCategoryIdRDD = payActionRDD.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
					@Override
					public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
						Row row = tuple._2;
						String payCategoryIds = row.getString(10);
						String[] payCategoryIdsSplited = payCategoryIds.split(",");
						List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
						for (String payCategoryId : payCategoryIdsSplited) {
							list.add(new Tuple2<Long, Long>(Long.valueOf(payCategoryId), 1L));
						}
						return list;
					}
				});

		return payCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
			@Override
			public Long call(Long v1, Long v2) throws Exception {
				return v1 + v2;
			}
		});
	}

	private static JavaPairRDD<Long, Long> getOrderCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD) {
		JavaPairRDD<String, Row> orderActionRDD = sessionid2detailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
			@Override
			public Boolean call(Tuple2<String, Row> tuple) throws Exception {
				Row row = tuple._2;
				return row.getString(8) != null;
			}
		});

		JavaPairRDD<Long, Long> orderCategoryIdRDD = orderActionRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
			@Override
			public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
				Row row = tuple._2;
				String orderCategoryIds = row.getString(8);
				String[] orderCategoryIdsSplited = orderCategoryIds.split(",");
				ArrayList<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
				for (String orderCategoryId : orderCategoryIdsSplited) {
					list.add(new Tuple2<Long, Long>(Long.valueOf(orderCategoryId), 1L));
				}
				return list;
			}
		});
		return orderCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
			@Override
			public Long call(Long v1, Long v2) throws Exception {
				return v1 + v2;
			}
		});
	}

	private static JavaPairRDD<Long, Long> getClickCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD) {
		JavaPairRDD<String, Row> clickActionRDD = sessionid2detailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
			@Override
			public Boolean call(Tuple2<String, Row> tuple) throws Exception {
				Row row = tuple._2;
				return row.get(6) != null;
			}
		});
		JavaPairRDD<Long, Long> clickCategoryIdRDD = clickActionRDD.mapToPair(new PairFunction<Tuple2<String, Row>, Long, Long>() {
			@Override
			public Tuple2<Long, Long> call(Tuple2<String, Row> tuple) throws Exception {
				long clickCategoryId = tuple._2.getLong(6);
				return new Tuple2<Long, Long>(clickCategoryId, 1L);
			}
		});

		return clickCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
			@Override
			public Long call(Long v1, Long v2) throws Exception {
				return v1 + v2;
			}
		});
	}

	private static void calculateAggrStat(String value, long taskid) {
		//从Accumulator统计串中获取值
		long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.SESSION_COUNT));

		long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_1s_3s));
		long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_4s_6s));
		long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_7s_9s));
		long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_10s_30s));
		long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_30s_60s));
		long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_1m_3m));
		long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_3m_10m));
		long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_10m_30m));
		long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.TIME_PERIOD_30m));

		long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_1_3));
		long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_4_6));
		long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_7_9));
		long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_10_30));
		long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_30_60));
		long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
				value, "\\|", Constants.STEP_PERIOD_60));

		// 计算各个访问时长和访问步长的范围
		double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
				(double) visit_length_1s_3s / (double) session_count, 2);
		double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
				(double) visit_length_4s_6s / (double) session_count, 2);
		double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
				(double) visit_length_7s_9s / (double) session_count, 2);
		double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
				(double) visit_length_10s_30s / (double) session_count, 2);
		double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
				(double) visit_length_30s_60s / (double) session_count, 2);
		double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
				(double) visit_length_1m_3m / (double) session_count, 2);
		double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
				(double) visit_length_3m_10m / (double) session_count, 2);
		double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
				(double) visit_length_10m_30m / (double) session_count, 2);
		double visit_length_30m_ratio = NumberUtils.formatDouble(
				(double) visit_length_30m / (double) session_count, 2);

		double step_length_1_3_ratio = NumberUtils.formatDouble(
				(double) step_length_1_3 / (double) session_count, 2);
		double step_length_4_6_ratio = NumberUtils.formatDouble(
				(double) step_length_4_6 / (double) session_count, 2);
		double step_length_7_9_ratio = NumberUtils.formatDouble(
				(double) step_length_7_9 / (double) session_count, 2);
		double step_length_10_30_ratio = NumberUtils.formatDouble(
				(double) step_length_10_30 / (double) session_count, 2);
		double step_length_30_60_ratio = NumberUtils.formatDouble(
				(double) step_length_30_60 / (double) session_count, 2);
		double step_length_60_ratio = NumberUtils.formatDouble(
				(double) step_length_60 / (double) session_count, 2);

		// 将统计结果封装为Domain对象
		SessionAggrStat sessionAggrStat = new SessionAggrStat();
		sessionAggrStat.setTaskid(taskid);
		sessionAggrStat.setSession_count(session_count);
		sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
		sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
		sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
		sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
		sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
		sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
		sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
		sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
		sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
		sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
		sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
		sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
		sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
		sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
		sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio);

		// 调用对应的DAO插入统计结果
		ISessionAggrStatDAO sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO();
		sessionAggrStatDAO.insert(sessionAggrStat);
	}

	private static void randomExtractSession(final long taskid, JavaPairRDD<String, String> sessionid2AggrInfoRDD,
	                                         JavaPairRDD<String, Row> sessionid2actionRDD) {
		/**
		 * 第一步:计算出每天每小时的session数量
		 */
		//获取<yyyy-MM-dd_HH,aggrInfo>格式的RDD
		JavaPairRDD<String, String> time2sessionidRDD = sessionid2AggrInfoRDD.mapToPair(
				new PairFunction<Tuple2<String, String>, String, String>() {
					@Override
					public Tuple2<String, String> call(Tuple2<String, String> tuple) throws Exception {
						String aggrInfo = tuple._2;
						String startTime = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_START_TIME);
						String dateHour = DateUtils.getDateHour(startTime);
						return new Tuple2<String, String>(dateHour, aggrInfo);
					}
				});
		//得到每天每小时session数量
		Map<String, Object> countMap = time2sessionidRDD.countByKey();

		/**
		 * 第二步:使用按时间比列随机抽取算法,计算出每天要抽取session的索引
		 */
		//将<yyyy-MM-dd_HH,count>格式的map,转换成<yyyy-MM-dd,<HH,count>>的格式
		HashMap<String, Map<String, Long>> dateHourCountMap = new HashMap<String, Map<String, Long>>();

		for (Map.Entry<String, Object> countEntry : countMap.entrySet()) {
			String dateHour = countEntry.getKey();
			String date = dateHour.split("_")[0];
			String hour = dateHour.split("_")[1];
			Long count = Long.valueOf(String.valueOf(countEntry.getValue()));

			Map<String, Long> hourCountMap = dateHourCountMap.get(date);
			if (MapUtils.isEmpty(hourCountMap)) {
				hourCountMap = new HashMap<String, Long>();
				dateHourCountMap.put(date, hourCountMap);
			}
			hourCountMap.put(hour, count);
		}

		//总共要抽取100个session,先按照天数进行平分
		int extractNumberPerDay = 100 / dateHourCountMap.size();

		//<date,<hour,(3,5,9,33)>>
		final HashMap<String, Map<String, List<Integer>>> dateHourExtractMap = new HashMap<String, Map<String, List<Integer>>>();

		//用于生成随机索引
		Random random = new Random(System.currentTimeMillis());

		for (Map.Entry<String, Map<String, Long>> dateHourCountEntry : dateHourCountMap.entrySet()) {
			String date = dateHourCountEntry.getKey();
			Map<String, Long> hourCountMap = dateHourCountEntry.getValue();

			//计算出这一天的session总数
			long sessionCount = 0;
			for (long hourCount : hourCountMap.values()) {
				sessionCount += hourCount;
			}

			Map<String, List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
			if (MapUtils.isEmpty(hourExtractMap)) {
				hourExtractMap = new HashMap<String, List<Integer>>();
				dateHourExtractMap.put(date, hourExtractMap);
			}

			//遍历每个小时数量
			for (Map.Entry<String, Long> hourCountEntry : hourCountMap.entrySet()) {
				String hour = hourCountEntry.getKey();
				long count = hourCountEntry.getValue();

				//计算每个小时的session数量,占据当天总session数量的比列,直接乘以每天要抽取的数量,就可以计算出每小时要抽取的session数量
				int hourExtractNumber = (int) (((double) count / (double) sessionCount) * extractNumberPerDay);
				if (hourExtractNumber > count) {
					hourExtractNumber = (int) count;
				}

				//获取当前小时存放的随机数list
				List<Integer> extractIndexList = hourExtractMap.get(hour);
				if (CollectionUtils.isEmpty(extractIndexList)) {
					extractIndexList = new ArrayList<Integer>();
					hourExtractMap.put(hour, extractIndexList);
				}

				//生成随机数
				for (int i = 0; i < hourExtractNumber; i++) {
					int extractIndex = random.nextInt((int) count);
					while (extractIndexList.contains(extractIndex)) {
						extractIndex = random.nextInt((int) count);
					}
					extractIndexList.add(extractIndex);
				}
			}
		}

		/**
		 * 第三步:遍历每天每小时的session,然后根据随机索引进行抽取
		 */
		//执行groupByKey,得到<dateHour,List<aggrInfo>>
		JavaPairRDD<String, Iterable<String>> time2sessionsRDD = time2sessionidRDD.groupByKey();

		JavaPairRDD<String, String> extractSessionidsRDD = time2sessionsRDD.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Iterable<String>>, String, String>() {
					@Override
					public Iterable<Tuple2<String, String>> call(Tuple2<String, Iterable<String>> tuple) throws Exception {
						List<Tuple2<String, String>> extractSessionids = new ArrayList<Tuple2<String, String>>();
						String dateHour = tuple._1;
						String date = dateHour.split("_")[0];
						String hour = dateHour.split("_")[1];
						Iterator<String> iterator = tuple._2.iterator();

						List<Integer> extractIndexList = dateHourExtractMap.get(date).get(hour);

						ISessionRandomExtractDAO sessionRandomExtractDAO = DAOFactory.getISessionRandomExtractDAO();

						int index = 0;
						while (iterator.hasNext()) {
							String sessionAggrInfo = iterator.next();

							//如果当前session是我们要的session
							if (extractIndexList.contains(index)) {
								String sessionid = StringUtils.getFieldFromConcatString(
										sessionAggrInfo, "\\|", Constants.FIELD_SESSION_ID);

								// 将数据写入MySQL
								SessionRandomExtract sessionRandomExtract = new SessionRandomExtract();
								sessionRandomExtract.setTaskid(taskid);
								sessionRandomExtract.setSessionid(sessionid);
								sessionRandomExtract.setStartTime(StringUtils.getFieldFromConcatString(
										sessionAggrInfo, "\\|", Constants.FIELD_START_TIME));
								sessionRandomExtract.setSearchKeywords(StringUtils.getFieldFromConcatString(
										sessionAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS));
								sessionRandomExtract.setClickCategoryIds(StringUtils.getFieldFromConcatString(
										sessionAggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS));

								sessionRandomExtractDAO.insert(sessionRandomExtract);

								// 将sessionid加入list
								extractSessionids.add(new Tuple2<String, String>(sessionid, sessionid));
							}
							index++;
						}
						return extractSessionids;
					}
				});
		/**
		 * 第四步:获取抽取出来的session明细数据
		 */
		JavaPairRDD<String, Tuple2<String, Row>> extractSessionDetailRDD = extractSessionidsRDD.join(sessionid2actionRDD);
		extractSessionDetailRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {
			@Override
			public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
				Row row = tuple._2._2;
				SessionDetail sessionDetail = new SessionDetail();
				sessionDetail.setTaskid(taskid);
				sessionDetail.setUserid(row.getLong(1));
				sessionDetail.setSessionid(row.getString(2));
				sessionDetail.setPageid(row.getLong(3));
				sessionDetail.setActionTime(row.getString(4));
				sessionDetail.setSearchKeyword(row.getString(5));
				sessionDetail.setClickCategoryId(row.getLong(6));
				sessionDetail.setClickProductId(row.getLong(7));
				sessionDetail.setOrderCategoryIds(row.getString(8));
				sessionDetail.setOrderProductIds(row.getString(9));
				sessionDetail.setPayCategoryIds(row.getString(10));
				sessionDetail.setPayProductIds(row.getString(11));

				ISessionDetailDAO sessionDetailDAO = DAOFactory.getISessionDetailDAO();
				sessionDetailDAO.insert(sessionDetail);
			}
		});
	}


	private static JavaPairRDD<String, Row> getSessionid2detailRDD(JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD,
	                                                               JavaPairRDD<String, Row> sessionid2actionRDD) {
		//过滤出来的数据和session全量数据join
		JavaPairRDD<String, Tuple2<String, Row>> filteredSessionidAggrInfoRDD = filteredSessionid2AggrInfoRDD.join(sessionid2actionRDD);

		JavaPairRDD<String, Row> sessionid2detailRDD = filteredSessionidAggrInfoRDD.mapToPair(
				new PairFunction<Tuple2<String, Tuple2<String, Row>>, String, Row>() {
					@Override
					public Tuple2<String, Row> call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
						return new Tuple2<String, Row>(tuple._1, tuple._2._2);
					}
				});
		return sessionid2detailRDD;
	}

	private static JavaPairRDD<String, String> filterSessionAndAggrStat(JavaPairRDD<String, String> sessionid2AggrInfoRDD,
	                                                                    JSONObject taskParam, final Accumulator<String> sessionAggrStatAccumulator) {
		//获取查询条件
		String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
		String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
		String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
		String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
		String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
		String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
		String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);

		String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
				+ (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
				+ (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
				+ (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
				+ (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
				+ (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
				+ (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");

		if (_parameter.endsWith("\\|")) {
			_parameter = _parameter.substring(0, _parameter.length() - 1);
		}

		final String parameter = _parameter;

		//进行筛选过滤
		JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = sessionid2AggrInfoRDD.filter(
				new Function<Tuple2<String, String>, Boolean>() {
					@Override
					public Boolean call(Tuple2<String, String> tuple) throws Exception {
						//获得聚合信息
						String aggrInfo = tuple._2;
						//按照年龄进行范围过滤
						if (!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
							return false;
						}
						//按照职业过滤
						if (!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL, parameter, Constants.PARAM_PROFESSIONALS)) {
							return false;
						}
						//按城市范围过滤
						if (!ValidUtils.in(aggrInfo, Constants.FIELD_CITY, parameter, Constants.PARAM_CITIES)) {
							return false;
						}
						//按性别过滤
						if (!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, parameter, Constants.PARAM_SEX)) {
							return false;
						}
						// 按照搜索词进行过滤
						if (!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS,
								parameter, Constants.PARAM_KEYWORDS)) {
							return false;
						}

						// 按照点击品类id进行过滤
						if (!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS,
								parameter, Constants.PARAM_CATEGORY_IDS)) {
							return false;
						}

						//到这一步都是通过过滤的数据,后面进行数据统计
						sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);
						Long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
						Long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH));
						calculateVisitLength(visitLength);
						calculateStepLength(stepLength);
						return true;
					}

					/**
					 * 计算访问时长范围
					 * @param visitLength
					 */
					private void calculateVisitLength(long visitLength) {
						if (visitLength >= 1 && visitLength <= 3) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
						} else if (visitLength >= 4 && visitLength <= 6) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
						} else if (visitLength >= 7 && visitLength <= 9) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
						} else if (visitLength >= 10 && visitLength <= 30) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
						} else if (visitLength > 30 && visitLength <= 60) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
						} else if (visitLength > 60 && visitLength <= 180) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
						} else if (visitLength > 180 && visitLength <= 600) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
						} else if (visitLength > 600 && visitLength <= 1800) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
						} else if (visitLength > 1800) {
							sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
						}
					}

					/**
					 * 计算访问步长范围
					 * @param stepLength
					 */
					private void calculateStepLength(long stepLength) {
						if (stepLength >= 1 && stepLength <= 3) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
						} else if (stepLength >= 4 && stepLength <= 6) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
						} else if (stepLength >= 7 && stepLength <= 9) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
						} else if (stepLength >= 10 && stepLength <= 30) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
						} else if (stepLength > 30 && stepLength <= 60) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
						} else if (stepLength > 60) {
							sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
						}
					}
				});
		return filteredSessionid2AggrInfoRDD;
	}

	private static JavaPairRDD<String, String> aggregateBySession(SQLContext sqlContext, JavaPairRDD<String, Row> sessionid2actionRDD) {
		//根据sessionid分组
		JavaPairRDD<String, Iterable<Row>> sessionid2ActionsRDD = sessionid2actionRDD.groupByKey();

		//转换成<userid,(sessionid,searchKeywords,clickCategoryIds,visitLength,stepLength,startTime)>,为了和user_info表join
		JavaPairRDD<Long, String> userid2PartAggrInfoRDD = sessionid2ActionsRDD.mapToPair(
				new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
					@Override
					public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
						String sessionid = tuple._1;
						Iterator<Row> iterator = tuple._2.iterator();
						//分组的session,搜索的哪些词
						StringBuffer searchKeywordsBuffer = new StringBuffer();
						//分组的session,点击了哪些品类
						StringBuffer clickCategoryidsBuffer = new StringBuffer();

						//用户id,同一个session,用户id肯定都相同
						Long userid = null;

						//session开始时间和结束时间
						Date startTime = null;
						Date endTime = null;

						//session的访问步长
						int stepLength = 0;

						while (iterator.hasNext()) {
							Row row = iterator.next();
							if (userid == null) {
								userid = row.getLong(1);
							}

							String searchKey = row.getString(5);
							Long clickCategoryid = row.getLong(6);

							//不为空且不包含,才加入
							if (StringUtils.isNotEmpty(searchKey) && !searchKeywordsBuffer.toString().contains(searchKey)) {
								searchKeywordsBuffer.append(searchKey + ",");
							}
							if (clickCategoryid != null && !clickCategoryidsBuffer.toString().contains(String.valueOf(clickCategoryid))) {
								clickCategoryidsBuffer.append(clickCategoryid + ",");
							}

							//计算session的开始时间和结束时间
							Date actionTime = DateUtils.parseTime(row.getString(4));
							if (startTime == null) {
								startTime = actionTime;
							}
							if (endTime == null) {
								endTime = actionTime;
							}
							if (actionTime.before(startTime)) {
								startTime = actionTime;
							}
							if (actionTime.after(endTime)) {
								endTime = actionTime;
							}

							//每次步长加1
							stepLength++;
						}

						//searchKeywordsBuffer和clickCategoryidsBuffer去掉尾部的逗号
						String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
						String clickCategoryids = StringUtils.trimComma(clickCategoryidsBuffer.toString());

						//计算session的访问时长
						long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;

						//对字符串进行拼接,格式为key=value|key=value
						String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionid + "|"
								+ Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|"
								+ Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryids + "|"
								+ Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
								+ Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
								+ Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);
						return new Tuple2<Long, String>(userid, partAggrInfo);
					}
				});

		//查询所有用户,并映射成<userid,Row>的格式
		String sql = "select * from user_info";
		JavaRDD<Row> userInfoRDD = sqlContext.sql(sql).javaRDD();

		JavaPairRDD<Long, Row> userid2InfoRDD = userInfoRDD.mapToPair(new PairFunction<Row, Long, Row>() {
			@Override
			public Tuple2<Long, Row> call(Row row) throws Exception {
				return new Tuple2<Long, Row>(row.getLong(0), row);
			}
		});

		//将session粒度聚合数据与用户信息进行join
		JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD = userid2PartAggrInfoRDD.join(userid2InfoRDD);

		//对join的数据进行处理,返回<sessionid,fullAggrInfo>格式的数据
		JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userid2FullInfoRDD.mapToPair(
				new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
					@Override
					public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple) throws Exception {
						//单个session关联的用户信息
						Row userInfoRow = tuple._2._2;
						//单个session关联的session聚合信息
						String partAggrInfo = tuple._2._1;
						//获取sessionid
						String sessionid = StringUtils.getFieldFromConcatString(partAggrInfo, "\\|", Constants.FIELD_SESSION_ID);

						int age = userInfoRow.getInt(3);
						String professional = userInfoRow.getString(4);
						String city = userInfoRow.getString(5);
						String sex = userInfoRow.getString(6);

						String fullAggrInfo = partAggrInfo + "|"
								+ Constants.FIELD_AGE + "=" + age + "|"
								+ Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
								+ Constants.FIELD_CITY + "=" + city + "|"
								+ Constants.FIELD_SEX + "=" + sex;
						return new Tuple2<String, String>(sessionid, fullAggrInfo);
					}
				});
		return sessionid2FullAggrInfoRDD;
	}

	private static JavaPairRDD<String, Row> getSessionid2ActionRDD(JavaRDD<Row> actionRDD) {
		return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
			@Override
			public Tuple2<String, Row> call(Row row) throws Exception {
				return new Tuple2<String, Row>(row.getString(2), row);
			}
		});
	}

	private static JavaRDD<Row> getActionRDDByDateRange(SQLContext sqlContext, JSONObject taskParam) {
		String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
		String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);
		String sql = "select * from user_visit_action where date >= '" + startDate + "' and date <= '" + endDate + "'";
		DataFrame df = sqlContext.sql(sql);
		return df.javaRDD();
	}

	public static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
		if (Boolean.valueOf(ConfigurationManager.getProperty(Constants.SPARK_LOCAL))) {
			MockData.mock(sc, sqlContext);
		}
	}
}
